package ru.gerunn.silent.core;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLServerSocketFactory;
import javax.net.ssl.SSLSocket;

import org.apache.log4j.Logger;

import ru.gerunn.silent.message.MessageQueue;
import ru.gerunn.silent.message.MessageQueueCallable;

public class Server {

	public static Logger log = Logger.getLogger(Server.class);

	private final int port;
	private boolean logToConsole = false;
	private final ConcurrentHashMap<String, Session> sessions;
	private final MessageQueue messageQueue;
	private final SSLServerSocket socket;

	public Server() throws ServerException {
		this(8883);
	}

	public Server(final int port) throws ServerException {
		this.port = port;
		this.sessions = new ConcurrentHashMap<String, Session>();
		this.messageQueue = new MessageQueue();

		SSLServerSocketFactory serverSocketFactory = (SSLServerSocketFactory) SSLServerSocketFactory.getDefault();
		try {
			this.socket = (SSLServerSocket) serverSocketFactory.createServerSocket(getPort());
			log.info("bound to port " + getPort());
		} catch (IOException e) {
			throw new ServerException("error while creating server socket", e);
		}
	}

	public boolean isLogToConsole() {
		return logToConsole;
	}

	public void setLogToConsole(boolean logToConsole) {
		this.logToConsole = logToConsole;
	}

	public ConcurrentHashMap<String, Session> getSessions() {
		return sessions;
	}

	public void removeSession(String id) {
		this.sessions.remove(id);
	}

	public MessageQueue getMessageQueue() {
		return messageQueue;
	}

	public int getPort() {
		return port;
	}

	public SSLServerSocket getSocket() {
		return socket;
	}

	public void start() throws ServerException {

		final ExecutorService pool = Executors.newCachedThreadPool();
		final MessageQueueCallable messageQueueCallable = new MessageQueueCallable(messageQueue, sessions);
		final Future<Object> messageQueueFuture = pool.submit(messageQueueCallable);
		log.info("message queue started");

		Callable<Object> acceptingCallable = new Callable<Object>() {
			public Object call() throws ServerException {
				SSLSocket newSocket = null;

				try {
					while (!messageQueueFuture.isDone() && (newSocket = (SSLSocket) socket.accept()) != null) {
						log.info("got socket");
						stub(newSocket, pool);
					}
				} catch (IOException e) {
					try {
						socket.close();
					} catch (IOException e1) {
						throw new ServerException(e1);
					}
					throw new ServerException("error while accepting new socket", e);
				}

				return null;
			}
		};

		final Future<Object> acceptingFuture = pool.submit(acceptingCallable);
		log.info("accepting sockets started");

		try {
			acceptingFuture.get();
			messageQueueFuture.get();
		} catch (InterruptedException e) {
			throw new ServerException(e);
		} catch (ExecutionException e) {
			throw new ServerException(e);
		}

		pool.shutdown();

	}

	public void stub(SSLSocket socket, ExecutorService pool) {
		User user = new User("User" + sessions.size());
		try {
			final Session session = new Session(socket, user, messageQueue);
			sessions.put(session.getId(), session);
			pool.submit(new Callable<Object>() {
				public Object call() throws SessionException {
					session.start();
					return null;
				}
			});
		} catch (SessionException e) {
			e.printStackTrace();
		}
		log.info("stub finished");
	}

}
